package com.haiyou.data.common.persister.mysql.persister;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;

import com.haiyou.common.util.date.DateUtilsV2;
import com.haiyou.common.util.string.StringFormatUitls;
import org.apache.commons.lang3.Validate;

import com.alibaba.fastjson2.util.TypeUtils;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.haiyou.common.entity.Entity;
import com.haiyou.common.spring.component.LifecycleableComponent;
import com.haiyou.common.util.collect.CollectionUtils;
import com.haiyou.common.util.collect.SortUtils;
import com.haiyou.common.util.mysql.BatchUtils;
import com.haiyou.common.util.object.ClassUtils;
import com.haiyou.data.common.persister.mysql.dao.BaseEntityDao;
import com.haiyou.data.common.persister.mysql.help.ColumnHelper;
import com.haiyou.data.common.persister.mysql.help.DatabaseHelper;
import com.haiyou.data.common.persister.mysql.help.EntityHelper;
import com.haiyou.data.common.persister.mysql.help.SqlHelper;
import com.haiyou.data.common.persister.mysql.help.TableHelper;
import com.haiyou.data.common.persister.mysql.meta.ColumnMeta;
import com.haiyou.data.common.persister.mysql.meta.DatabaseMeta;
import com.haiyou.data.common.persister.mysql.meta.TableMeta;
import com.haiyou.data.common.persister.mysql.annotation.Table.RollPolicy;
import com.haiyou.data.common.persister.mysql.annotation.Index.IndexPolicy;


import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.ObjectUtils;

/**
 * 持久器
 * 
 * @author Administrator
 *
 * @param <Id>
 * @param <E>
 */
@Slf4j
public abstract class Persister<Id extends Serializable & Comparable<Id>, E extends Entity<Id>>
		implements LifecycleableComponent, BaseEntityDao<Id, E> {

	public static final Map<Class<? extends Entity<?>>, Persister<?, ?>> persisterPool = Maps.newHashMap();

	public static <P extends Persister<?, ?>> P getInstance(Class<? extends Entity<?>> entityClass) {
		P instance = (P) persisterPool.get(entityClass);
		Validate.isTrue(instance != null, "[%s]的持久器不存在", entityClass);
		return instance;
	}

	protected Class<Id> idClass;// 实体键类型

	protected Class<E> entityClass;// 实体值类型

	@Getter
	protected DatabaseMeta databaseMeta;// 数据库信息

	@Getter
	protected TableMeta tableMeta;// 数据表信息

	@Override
	public Class<Id> getIdClass() {
		Class<Id> clz = getIdClass(this.getClass());
		Objects.requireNonNull(clz, this.getClass().getName());
		return clz;
	}

	private Class<Id> getIdClass(Class<?> clazz) {
		Class<?>[] cls = ClassUtils.getTypeArguments(clazz);
		Class<Id> clz = (Class<Id>) (cls.length == 2 ? cls[0] : null);
		if (clz != null)
			return clz;
		return getIdClass(clazz.getSuperclass());
	}

	@Override
	public Class<E> getEntityClass() {
		Class<?>[] cls = ClassUtils.getTypeArguments(this.getClass());
		Class<E> clz = (Class<E>) (cls.length == 2 ? cls[1] : cls[0]);
		Objects.requireNonNull(clz, this.getClass().getName());
		return clz;
	}

	@Override
	public int getOrder() {
		return persister_order;// TODO 确保一定要在Accessor之前初始化
	}

	@Override
	public void init() {
		idClass = getIdClass();
		entityClass = getEntityClass();
		databaseMeta = DatabaseHelper.fix(entityClass);
		if (databaseMeta == null) {
			return;
		}
		tableMeta = TableHelper.fix(entityClass);
		Persister<?, ?> old = persisterPool.putIfAbsent(entityClass, this);
		Validate.isTrue(old == null, "重复添加持久器,entityClass=[%s],persister=[%s],old=[%s]", entityClass, this, old);
	}

	@Override
	public void destory() {

	}

	@Override
	public int sqlBatchNum() {
		return tableMeta == null ? BaseEntityDao.super.sqlBatchNum() : tableMeta.getSqlBatchNum();
	}

	protected void checkTableIfDoesExist() {
		if (tableMeta.getRollPolicy() != RollPolicy.NONE) {
			String db = DatabaseHelper.getOnlyReallyName(databaseMeta);
			String tab = RollPolicy.fixName(tableMeta.getSimpleName(), tableMeta.getRollPolicy());
			if (!TableHelper.existsTable(db, tab)) {
				TableHelper.createTableIfNotExists(tableMeta, db, tab);
			}
		}
	}

	protected String parseDatabase(Id id) {
		return databaseMeta.getCluster() > 1 ? DatabaseHelper.findDatabaseByPk(databaseMeta, id)
				: DatabaseHelper.getOnlyReallyName(databaseMeta);
	}

	protected String parseTable(Id id) {
		return tableMeta.getCluster() > 1 ? TableHelper.findTableByPk(tableMeta, id)
				: tableMeta.getRollPolicy() != RollPolicy.NONE
						? RollPolicy.fixName(tableMeta.getSimpleName(), tableMeta.getRollPolicy())
						: tableMeta.getSimpleName();
	}

	protected void checkRollAndIncrTableUpdate() {
		if (tableMeta.getRollPolicy() != RollPolicy.NONE && TableHelper.pkAutoIncr(tableMeta))
			throw new RuntimeException("无法使用主键更新，因为表结构是滚表并且自增主键:" + entityClass);
	}

	/**
	 * 转换成持久化的值
	 * 
	 * @param columnMeta
	 * @param entity
	 * @return
	 */
	protected Object cast(ColumnMeta columnMeta, Entity<Id> entity) {
		return ColumnHelper.cast(columnMeta, entity);
	}

	@Override
	public boolean insert(E entity) {
		checkTableIfDoesExist();
		String db = parseDatabase(entity.getId());
		String tab = parseTable(entity.getId());
		return executeUpdate(db, SqlHelper.insert(entity, db, tab, !TableHelper.pkAutoIncr(tableMeta))) == 1;
	}

	@Override
	public void insert(Iterable<? extends E> iterable) {
		checkTableIfDoesExist();
		Map<String, Set<String>> sqlMap = Maps.newHashMap();
		iterable.forEach(entity -> {
			String db = parseDatabase(entity.getId());
			String tab = parseTable(entity.getId());
			sqlMap.computeIfAbsent(db, k -> Sets.newHashSet())
					.add(SqlHelper.insert(entity, db, tab, !TableHelper.pkAutoIncr(tableMeta)));
		});
		sqlMap.forEach((db, sqls) -> executeBatch(db, sqls));
	}

	@Override
	public boolean insertOrUpdate(E entity) {

		checkRollAndIncrTableUpdate();

		String db = parseDatabase(entity.getId());
		String tab = parseTable(entity.getId());
		
		return executeUpdate(db, SqlHelper.insertOnDuplicateKeyUpdate(entity, db, tab)) == 1;
	}

	@Override
	public boolean insertOrUpdateExcludeAutoId(E entity) {

//		checkRollAndIncrTableUpdate();

		checkTableIfDoesExist();

		String db = parseDatabase(entity.getId());
		String tab = parseTable(entity.getId());

		return executeUpdate(db, SqlHelper.insertOnDuplicateKeyUpdateExcludeAutoId(entity, db, tab)) == 1;
	}

	@Override
	public void insertOrUpdate(Iterable<? extends E> iterable) {

		checkRollAndIncrTableUpdate();
		Map<String, Set<String>> sqlMap = Maps.newHashMap();
		iterable.forEach(entity -> {
			String db = parseDatabase(entity.getId());
			String tab = parseTable(entity.getId());
			sqlMap.computeIfAbsent(db, k -> Sets.newHashSet())
					.add(SqlHelper.insertOnDuplicateKeyUpdate(entity, db, tab));
		});
		sqlMap.forEach((db, sqls) -> executeBatch(db, sqls));
	}

	@Override
	public boolean insertIfNotExists(E entity) {

		checkRollAndIncrTableUpdate();

		String db = parseDatabase(entity.getId());
		String tab = parseTable(entity.getId());
		return executeUpdate(db, SqlHelper.insertIfNotExists(entity, db, tab)) == 1;
	}

	@Override
	public void insertIfNotExists(Iterable<? extends E> iterable) {

		checkRollAndIncrTableUpdate();

		Map<String, Set<String>> sqlMap = Maps.newHashMap();
		iterable.forEach(entity -> {
			String db = parseDatabase(entity.getId());
			String tab = parseTable(entity.getId());
			sqlMap.computeIfAbsent(db, k -> Sets.newHashSet()).add(SqlHelper.insertIfNotExists(entity, db, tab));
		});
		sqlMap.forEach((db, sqls) -> executeBatch(db, sqls));
	}

	@Override
	public boolean replace(E entity) {

		checkRollAndIncrTableUpdate();

		String db = parseDatabase(entity.getId());
		String tab = parseTable(entity.getId());
		return executeUpdate(db, SqlHelper.replace(entity, db, tab)) == 1;
	}

	@Override
	public void replace(Iterable<? extends E> iterable) {

		checkRollAndIncrTableUpdate();

		Map<String, Set<String>> sqlMap = Maps.newHashMap();
		iterable.forEach(entity -> {
			String db = parseDatabase(entity.getId());
			String tab = parseTable(entity.getId());
			sqlMap.computeIfAbsent(db, k -> Sets.newHashSet()).add(SqlHelper.update(entity, db, tab));
		});
		sqlMap.forEach((db, sqls) -> executeBatch(db, sqls));
	}

	/** 指定库,表,插入并得到自增主键的值 */
	public Integer return_generated_Id(String db, String tab, E entity) {

		Validate.isTrue(TableHelper.pkAutoIncr(tableMeta), "非法操作[%s]", tableMeta);
		Validate.isTrue(entity.getId() == null, "非法操作[%s]", entity);

		Stopwatch stopwatch = Stopwatch.createStarted();
		String sql = null;
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		ResultSet resultSet = null;
		try {
			connection = on(db);
			preparedStatement = connection.prepareStatement(sql = SqlHelper.insert(entity, db, tab, false),
					PreparedStatement.RETURN_GENERATED_KEYS);
			int r = preparedStatement.executeUpdate();
			if (r > 0) {
				resultSet = preparedStatement.getGeneratedKeys();
				if (resultSet != null && resultSet.next()) {
					int id = resultSet.getInt(1);
					return Integer.valueOf(id);
				}
			}
			return null;
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sql, e);
		} finally {
			off(resultSet, preparedStatement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > executeSlowly())
				log.warn("SLOWLY: [{}] elapsed [{} ms]", sql, elapsed);
			else
				log.debug("[{}] elapsed [{} ms]", sql, elapsed);
		}
	}

	@Override
	public Integer return_generated_Id(E entity) {
		Validate.isTrue(databaseMeta.getCluster() == 1, "非法操作[%s]", databaseMeta);
		Validate.isTrue(tableMeta.getCluster() == 1, "非法操作[%s]", tableMeta);
		Validate.isTrue(tableMeta.getRollPolicy() == RollPolicy.NONE, "非法操作[%s]", tableMeta);
		return return_generated_Id(DatabaseHelper.getOnlyReallyName(databaseMeta), tableMeta.getSimpleName(), entity);
	}

	@Override
	public boolean delete(Id id) {

		checkRollAndIncrTableUpdate();

		String pk = TableHelper.getPkName(tableMeta);
		String db = parseDatabase(id);
		if (tableMeta.getRollPolicy() != RollPolicy.NONE) {
			for (String tab : TableHelper.allTables(tableMeta))
				executeUpdate(db, String.format("delete from `%s`.`%s` where `%s`='%s'", db, tab, pk, id));
			return true;
		}
		String tab = parseTable(id);
		return executeUpdate(db, String.format("delete from `%s`.`%s` where `%s`='%s'", db, tab, pk, id)) == 1;
	}

	@Override
	public void delete(Iterable<? extends Id> ids) {

		checkRollAndIncrTableUpdate();

		String pk = TableHelper.getPkName(tableMeta);
		DatabaseHelper.matchedDatabases(databaseMeta, ids)
				.forEach((db, dbids) -> TableHelper.matchedTables(tableMeta, dbids)
						.forEach((tab, tabids) -> executeUpdate(db, SqlHelper.delete(db, tab, pk, tabids))));
	}

	@Override
	public void deleteByCondition(String condition) {
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta))
				executeUpdate(db, String.format("delete from `%s`.`%s` where %s", db, tab, condition));
		}
	}

	@Override
	public void delete() {
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta))
				executeUpdate(db, String.format("delete from `%s`.`%s`", db, tab));
		}
	}

	@Override
	public void truncate() {
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta))
				executeUpdate(db, String.format("truncate  `%s`.`%s`", db, tab));
		}
	}

	@Override
	public boolean update(E entity) {

		checkRollAndIncrTableUpdate();

		String db = parseDatabase(entity.getId());
		if (tableMeta.getRollPolicy() != RollPolicy.NONE) {
			for (String tab : TableHelper.allTables(tableMeta))
				executeUpdate(db, SqlHelper.update(entity, db, tab));
			return true;
		}
		String tab = parseTable(entity.getId());
		return executeUpdate(db, SqlHelper.update(entity, db, tab)) == 1;
	}

	@Override
	public void update(Iterable<? extends E> iterable) {

		checkRollAndIncrTableUpdate();

		Map<String, Set<String>> sqlMap = Maps.newHashMap();
		iterable.forEach(entity -> {
			String db = parseDatabase(entity.getId());
			if (tableMeta.getRollPolicy() != RollPolicy.NONE) {
				for (String tab : TableHelper.allTables(tableMeta))
					sqlMap.computeIfAbsent(db, k -> Sets.newHashSet()).add(SqlHelper.update(entity, db, tab));
			} else {
				String tab = parseTable(entity.getId());
				sqlMap.computeIfAbsent(db, k -> Sets.newHashSet()).add(SqlHelper.update(entity, db, tab));
			}
		});
		sqlMap.forEach((db, sqls) -> executeBatch(db, sqls));
	}

	@Override
	public boolean contains(Id id) {
		String db = parseDatabase(id);
		if (tableMeta.getRollPolicy() != RollPolicy.NONE) {
			for (String tab : TableHelper.allTables(tableMeta)) {
				if (contains(db, tab, id))
					return true;
			}
			return false;
		}
		String tab = parseTable(id);
		return contains(db, tab, id);
	}

	public boolean contains(String db, String tab, Id id) {
		String pk = TableHelper.getPkName(tableMeta);
		String sql = String.format("select `%s` from `%s`.`%s` where `%s` ='%s'", pk, db, tab, pk, id);
		AtomicReference<Id> atomicReference = new AtomicReference<>();
		executeQuery(db, sql, rs -> {
			try {
				Object v = rs.getObject(1);
				if (v != null) {
					Id x = TypeUtils.cast(v, idClass);
					atomicReference.set(x);
				}
			} catch (Exception e) {
				throw new RuntimeException(e.getMessage() + ":" + sql, e);
			}
		});
		return atomicReference.get() != null;
	}

	@Override
	public E query(Id id) {

		checkRollAndIncrTableUpdate();

		String db = parseDatabase(id);
		List<Map<String, Object>> list = null;
		if (tableMeta.getRollPolicy() != RollPolicy.NONE) {
			for (String tab : TableHelper.allTables(tableMeta)) {
				list = executeQuery(db, SqlHelper.query(entityClass, db, tab, id));
				if (!CollectionUtils.isEmpty(list))
					return EntityHelper.wrapped(entityClass, list.get(0), tableMeta);
			}
			return null;
		}
		String tab = parseTable(id);
		list = executeQuery(db, SqlHelper.query(entityClass, db, tab, id));
		if (!CollectionUtils.isEmpty(list))
			return EntityHelper.wrapped(entityClass, list.get(0), tableMeta);
		return null;
	}

	@Override
	public List<E> queryBatch(Iterable<? extends Id> ids) {
		checkRollAndIncrTableUpdate();
		List<Map<String, Object>> list = Lists.newArrayList();
		DatabaseHelper.matchedDatabases(databaseMeta, ids)
				.forEach((db, dbids) -> TableHelper.matchedTables(tableMeta, dbids)
						.forEach((tab, tabids) -> list.addAll(executeQueryBatchFromTable(db, tab, tabids))));
		if (CollectionUtils.isEmpty(list))
			return Collections.emptyList();
		List<E> result = Lists.newArrayList();
		for (Map<String, Object> map : list) {
			E entity = EntityHelper.wrapped(entityClass, map, tableMeta);
			result.add(entity);
		}
		return result;
	}

	public List<Map<String, Object>> executeQueryBatchFromTable(String db, String tab,
			Iterable<? extends Serializable> ids) {

		if (Iterables.size(ids) <= sqlBatchNum())
			return executeQuery(db, SqlHelper.queryBatch(entityClass, db, tab, ids));

		List<Map<String, Object>> result = Lists.newArrayListWithCapacity(sqlBatchNum());
		BatchUtils.toLists(ids, sqlBatchNum())
				.forEach(list -> result.addAll(executeQuery(db, SqlHelper.queryBatch(entityClass, db, tab, list))));

		return result;
	}

	@Override
	public List<E> queryByCondition(String condition) {
		List<E> result = Lists.newArrayList();
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta)) {
				List<Map<String, Object>> list = executeQuery(db,
						String.format("select * from `%s`.`%s` where %s ", db, tab, condition));
				for (Map<String, Object> map : list) {
					E entity = EntityHelper.wrapped(entityClass, map, tableMeta);
					result.add(entity);
				}
			}
		}
		return result;
	}

	public List<E> queryByCondition(String condition, Date startTime, Date endDate) {
		List<E> result = new LinkedList<>();
		if (startTime == null || endDate == null) {
			List<String> tableNames = Lists.newArrayList(TableHelper.allTables(tableMeta));
			tableNames = sortTableNames(tableNames);
			for (String db : databaseMeta.getReallyNames()) {
				for (String tableName : tableNames) {
					List<Map<String, Object>> list = executeQuery(db,
							String.format("select * from `%s`.`%s` where %s ", db, tableName, condition));
					log.info("sql:{},size：{}", tableName, list.size());
					for (Map<String, Object> map : list) {
						E entity = EntityHelper.wrapped(entityClass, map, tableMeta);
						result.add(entity);
					}
				}
			}
			return result;
		}
		final String prefix = "@_";
		Calendar calendar = Calendar.getInstance();
		Date date = new Date(endDate.getTime());
		do {
			String tableName = prefix + tableMeta.getSimpleName() + "_" + DateUtilsV2.parseYYYYMMDD(date.getTime());
			for (String db : databaseMeta.getReallyNames()) {
				String sql = String.format("select * from `%s`.`%s` where %s ", db, tableName, condition);
				List<Map<String, Object>> list = executeQuery(db, sql);
				log.info("sql:{},size：{}", tableName, list.size());
				for (Map<String, Object> map : list) {
					E entity = EntityHelper.wrapped(entityClass, map, tableMeta);
					result.add(entity);
				}
			}
			calendar.setTime(date);
			calendar.add(Calendar.DATE, -1);
			date = calendar.getTime();
		} while (date.after(startTime));
		return result;
	}

	/**
	 * 表名进行排序
	 * @param tableNames
	 * @return
	 */
	public List<String> sortTableNames(List<String> tableNames){
		if(ObjectUtils.isEmpty(tableNames)) {
			return Lists.newArrayList();
		}
		String tableName = tableMeta.getSimpleName();
		String format = RollPolicy.PREFIX + "_{}_";
		String suffix = StringFormatUitls.format(format, tableName);
		Map<Integer, String> names = Maps.newHashMap();
		List<Integer> sorts = Lists.newArrayList();
		for (String name : tableNames) {
			try {
				int date = Integer.parseInt(name.replaceFirst(suffix, ""));
				sorts.add(date);
				names.put(date, name);
			} catch (NumberFormatException e) {
				log.error("{} {} 转换异常",name,suffix);
			}
		}
		sorts.sort(Collections.reverseOrder());
		List<String> values = Lists.newArrayList();
		for (Integer sort : sorts) {
			values.add(names.get(sort));
		}
		return values;
	}

	public List<E> queryByConditionLimits(String condition, int limitStart, int limitEnd) {
		int start = Math.min(limitStart, limitEnd);
		int end = Math.max(limitStart, limitEnd);
		List<E> result = Lists.newArrayList();
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta))
				result.addAll(queryByConditionLimits(db, tab, condition, start, end));
		}
		return result;
	}

	public List<E> queryByConditionLimits(String db, String tab, String condition, int startIndex, int endIndex) {

		startIndex = Math.min(startIndex, endIndex);
		endIndex = Math.max(startIndex, endIndex);

		if(startIndex < endIndex) {
			endIndex = endIndex - startIndex;
		}
		
		List<Map<String, Object>> list = Lists.newArrayList();
		list.addAll(executeQuery(db, String.format("select *from `%s`.`%s` where %s limit %d,%d", db, tab, condition,
				startIndex, endIndex)));
		if (CollectionUtils.isEmpty(list))
			return Collections.emptyList();
		List<E> result = Lists.newArrayList();
		for (Map<String, Object> map : list) {
			E entity = EntityHelper.wrapped(entityClass, map, tableMeta);
			result.add(entity);
		}
		return result;
	}

	public String multiValue(Object... values) {
		StringBuilder sb = new StringBuilder();
		for (int i = 0; i < values.length; i++) {
			sb.append(values[i]);
			if (i < values.length - 1)
				sb.append(' ');
		}
		return sb.toString();
	}

	@Override
	public List<E> queryByFullText(String[] columns, Object[] params, boolean useBooleanMode) {
		String sql = "select *from `%s`.`%s` where match(%s) against('%s'%s)";
		List<Map<String, Object>> list = Lists.newArrayList();
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta)) {
				List<Map<String, Object>> listMap = executeQuery(db,
						String.format(sql, db, tab, IndexPolicy.multiColumn(columns), multiValue(params),
								!useBooleanMode ? "" : " in boolean mode"));
				list.addAll(listMap);
			}
		}
		if (CollectionUtils.isEmpty(list))
			return Collections.emptyList();
		List<E> result = Lists.newArrayList();
		for (Map<String, Object> map : list) {
			E entity = EntityHelper.wrapped(entityClass, map, tableMeta);
			result.add(entity);
		}
		return result;
	}

	public List<Map<String, Object>> executeQueryLimitsFromTable(String db, String tab, int startIndex, int endIndex,
			boolean reversed) {
		startIndex = Math.min(startIndex, endIndex);
		endIndex = Math.max(startIndex, endIndex);
		
		if(startIndex < endIndex) {
			endIndex = endIndex - startIndex;
		}
		
		if (reversed)
			return executeQuery(db, String.format("select *from `%s`.`%s` order by `%s` desc limit %d,%d", db, tab,
					TableHelper.getPkName(tableMeta), startIndex, endIndex));
		else
			return executeQuery(db, String.format("select *from `%s`.`%s` limit %d,%d", db, tab, startIndex, endIndex));
	}

	
	

	
	
	@Override
	public List<E> queryAll() {
		List<Map<String, Object>> list = Lists.newArrayList();
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta)) {
				int total = (int) count(db, tab);
				if (total < sqlBatchNum()) {
					list.addAll(executeQueryLimitsFromTable(db, tab, 0, total, false));
				} else {
					int sqlBatchNum = sqlBatchNum();
					int num = total % sqlBatchNum == 0 ? total / sqlBatchNum : total / sqlBatchNum + 1;
					for (int i = 0; i < num; i++)
						list.addAll(executeQueryLimitsFromTable(db, tab, i == 0 ? 0 : (sqlBatchNum * i),
								i == 0 ? sqlBatchNum : sqlBatchNum * (i + 1), false));
				}
			}
		}
		if (CollectionUtils.isEmpty(list))
			return Collections.emptyList();
		List<E> result = Lists.newArrayList();
		for (Map<String, Object> map : list) {
			E entity = EntityHelper.wrapped(entityClass, map, tableMeta);
			result.add(entity);
		}
		return result;
	}

	public List<E> queryByLimits(int startIndex, int endIndex, boolean reversed) {

		startIndex = Math.min(startIndex, endIndex);
		endIndex = Math.max(startIndex, endIndex);

		List<E> result = Lists.newArrayList();
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta))
				result.addAll(queryByLimits(db, tab, startIndex, endIndex, reversed));
		}
		return result;
	}

	public List<E> queryByLimits(String db, String tab, int startIndex, int endIndex, boolean reversed) {

		startIndex = Math.min(startIndex, endIndex);
		endIndex = Math.max(startIndex, endIndex);

		List<Map<String, Object>> list = Lists.newArrayList();
		list.addAll(executeQueryLimitsFromTable(db, tab, startIndex, endIndex, reversed));
		if (CollectionUtils.isEmpty(list))
			return Collections.emptyList();
		List<E> result = Lists.newArrayList();
		for (Map<String, Object> map : list) {
			E entity = EntityHelper.wrapped(entityClass, map, tableMeta);
			result.add(entity);
		}
		return result;
	}

	public Set<Id> queryIds(String db, String tab) {
		Set<Id> ids = Sets.newHashSet();
		String sql = String.format("select `%s` from `%s`.`%s`", TableHelper.getPkName(tableMeta), db, tab);
		executeQuery(db, sql, rs -> {
			try {
				Object v = rs.getObject(1);
				if (v != null) {
					Id x = TypeUtils.cast(v, idClass);
					ids.add(x);
				}
			} catch (Exception e) {
				throw new RuntimeException(e.getMessage() + ":" + sql, e);
			}
		});
		return ids;
	}

	@Override
	public List<Id> queryIds() {
		List<Id> ids = Lists.newArrayList();
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta))
				ids.addAll(queryIds(db, tab));
		}
		return ids;
	}

	public int count(String db, String tab) {
		checkTableIfDoesExist();
		LongAdder adder = new LongAdder();
		String sql = String.format("select count(*) from `%s`.`%s`", db, tab);
		executeQuery(db, sql, rs -> {
			try {
				Long count = rs.getLong(1);
				adder.add(count);
			} catch (Exception e) {
				throw new RuntimeException(e.getMessage() + ":" + sql, e);
			}
		});
		return adder.intValue();
	}

	@Override
	public int count() {
		LongAdder adder = new LongAdder();
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta))
				adder.add(count(db, tab));
		}
		return adder.intValue();
	}

	public int countByCondition(String db, String tab, String condition) {
		LongAdder adder = new LongAdder();
		String sql = String.format("select count(*) from `%s`.`%s` where %s", db, tab, condition);
		executeQuery(db, sql, rs -> {
			try {
				Long count = rs.getLong(1);
				adder.add(count);
			} catch (Exception e) {
				throw new RuntimeException(e.getMessage() + ":" + sql, e);
			}
		});
		return adder.intValue();
	}

	@Override
	public int countByCondition(String condition) {
		LongAdder adder = new LongAdder();
		for (String db : databaseMeta.getReallyNames()) {
			for (String tab : TableHelper.allTables(tableMeta))
				adder.add(countByCondition(db, tab, condition));
		}
		return adder.intValue();
	}

	public Id maxPk(String db, String tab) {
		List<Id> kList = Lists.newArrayList();
		String sql = String.format("select max(`%s`) from `%s`.`%s`", TableHelper.getPkName(tableMeta), db, tab);
		executeQuery(db, sql, rs -> {
			try {
				Object v = rs.getObject(1);
				if (v != null) {
					Id x = TypeUtils.cast(v, idClass);
					kList.add(x);
				}
			} catch (Exception e) {
				throw new RuntimeException(e.getMessage() + ":" + entityClass + "," + db + "," + tab + "," + sql, e);
			}
		});
		if (!CollectionUtils.isEmpty(kList)) {
			SortUtils.quickSort(kList, 0, kList.size() - 1);
			Id max = kList.get(kList.size() - 1);
			return max;
		}
		return null;
	}

	@Override
	public Id maxPk() {
		List<Id> kList = Lists.newArrayList();
		databaseMeta.getReallyNames().forEach(db -> {
			TableHelper.allTables(tableMeta).forEach(tab -> {
				Id max = maxPk(db, tab);
				if (max != null)
					kList.add(max);
			});
		});
		if (!CollectionUtils.isEmpty(kList)) {
			SortUtils.quickSort(kList, 0, kList.size() - 1);
			Id max = kList.get(kList.size() - 1);
			return max;
		}
		return null;
	}


	public LinkedHashMap<String, Integer> mapByCondition(String condition) {
		LinkedHashMap<String, Integer> map = new LinkedHashMap<>();
		for (String db : databaseMeta.getReallyNames()) {
			Set<String> set = TableHelper.allTables(tableMeta);
			ArrayList<String> list = new ArrayList<>(set);
			Collections.sort(list);
			for (String tab : list)
				map.put(tab, countByCondition(db, tab, condition));
		}
		return map;
	}

}
